package weaver.splittable.biz.workflow;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import weaver.conn.ConnectionPool;
import weaver.conn.WeaverConnection;
import weaver.conn.WeaverPreparedStatement;
import weaver.conn.WeaverStatement;
import weaver.splittable.STConfig;
import weaver.splittable.biz.STBizDataCleaner;
import weaver.splittable.db.STDBConstants;
import weaver.splittable.db.STDBSupport;
import weaver.splittable.task.STSyncTask;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 流程数据清理器
 * 按规则清理流程现有表数据 超过分表时间 并且已经归档的数据
 * author: YUEYANG 2023-01-16
 */
public class STWorkflowDataCleaner implements STBizDataCleaner {

    private static final Logger LOGGER = Logger.getLogger(STConfig.LOGGER_NAME);

    private static final String WORKFLOW_REQUESTBASE_DEL_ORACLE = "delete from %s where rownum<=1000 and createdate <= '%s' and currentnodetype = '3'";
    private static final String WORKFLOW_REQUESTBASE_DEL_MYSQL = "delete from %s where createdate <= '%s' and currentnodetype = '3' limit 1000";
    private static final String WORKFLOW_REQUESTBASE_DEL_SQLSERVER = "delete top(1000) from %s where createdate < '%s' and currentnodetype = '3'";

    private static final String WORKFLOW_CURRENTOPERATOR_DEL_ORACLE = "delete from %s where rownum<=1000 and receivedate <= '%s'";
    private static final String WORKFLOW_CURRENTOPERATOR_DEL_MYSQL = "delete from %s where receivedate <= '%s' limit 1000";
    private static final String WORKFLOW_CURRENTOPERATOR_DEL_SQLSERVER = "delete top(1000) from %s where receivedate <= '%s'";

    private static final String WORKFLOW_REQUESTLOG_DEL_ORACLE = "delete from %s where rownum<=1000 and operatedate <= '%s'";
    private static final String WORKFLOW_REQUESTLOG_DEL_MYSQL = "delete from %s where operatedate <= '%s' limit 1000";
    private static final String WORKFLOW_REQUESTLOG_DEL_SQLSERVER = "delete top(1000) from %s where operatedate <= '%s'";

    private static final String OTHER_TABLE_DEL_SEL_ORACLE = "delete from %s where rownum<=1000 and createdate <= '%s' and currentnodetype='3'";
    private static final String OTHER_TABLE_DEL_SEL_MYSQL = "delete from %s a LEFT JOIN workflow_requestbase b ON a.REQUESTID = b.REQUESTID WHERE b.REQUESTID is NULL AND  a.REQUESTID <= %s limit 1000";
    private static final String OTHER_TABLE_DEL_SEL_SQLSERVER = "delete from %s WHERE requestid in ( select top 1000 requestid from %s a LEFT JOIN workflow_requestbase b ON a.REQUESTID = b.REQUESTID WHERE b.REQUESTID is NULL AND  a.REQUESTID <= %s )";

    /**
     * 1. 清理前进行一次数据同步
     * 2. 将要清理的数据插入 数据清理表 中
     * 3. 执行清理
     */
    @Override
    public void clear() {
        Long beginClearTime = System.currentTimeMillis();
        for (Map.Entry<String, String> entry : STWorkflowConstants.SPLIT_TABLES.entrySet()) {
            String tableName = entry.getKey();
            String dataIdColumn = entry.getValue();
            String clearSql = getClearSql(tableName);
            int clearCounts = 0;
            boolean isFinished = false;
            while (!isFinished) {

                clearCounts = 0;

                Long startTime = System.currentTimeMillis();

                WeaverConnection connection = null;
                WeaverStatement statement = null;
                try {
                    connection = ConnectionPool.getInstance().getConnection();
                    connection.setAutoCommit(false);
                    statement = (WeaverStatement) connection.createStatement();

                    // 清理前同步一次数据
                    STSyncTask syncTask = new STSyncTask();
                    syncTask.onceSync(connection);

                    // 将要清理的数据插入 数据清理表 中并返回要删除的数据ID
                    Set<String> deleteDataIds = saveClearData(connection, clearSql, tableName, dataIdColumn);
                    clearCounts = deleteDataIds.size();
                    if (clearCounts > 0) {
                        String deleteIdStr = StringUtils.join(deleteDataIds, ",");
                        String deleteSql = "delete from " + tableName + " where " + dataIdColumn + " in (" + deleteIdStr + ")";

                        // 执行清理
                        clearCounts = statement.originExecuteUpdate(deleteSql);

                        if (clearCounts < 1000) {
                            isFinished = true;
                        }
                    } else {
                        isFinished = true;
                    }

                    connection.commit();
                } catch (Exception e) {
                    try {
                        connection.rollback();
                    } catch (SQLException ex) {
                        LOGGER.error("STWorkflowDataCleaner clear rollback error:", e);
                    }
                    LOGGER.error("STWorkflowDataCleaner clear error:", e);
                } finally {
                    if (statement != null) {
                        statement.closeNoException();
                    }
                    if (connection != null) {
                        try {
                            connection.setAutoCommit(true);
                        } catch (SQLException e) {
                            LOGGER.error("STWorkflowDataCleaner clear setAutoCommit true error:", e);
                        }
                        ConnectionPool.getInstance().returnConnection(connection);
                    }
                }

                Long endTime = System.currentTimeMillis();
                LOGGER.info("STWorkflowDataCleaner clear " + tableName + " count: " + clearCounts + ", time: " + (endTime - startTime) + "ms");

                // 如果清理任务总执行时间超过4个小时 则终止本次清理
                Long runningTime = endTime - beginClearTime;
                if (runningTime > 4 * 60 * 60 * 1000) {
                    LOGGER.info("STWorkflowDataCleaner running time over 4 hours, stop clear task!");
                    isFinished = true;
                }

            }
        }

        //清理完更新阈值
        updateRequestIdThresholdToDB();
    }
    private void updateRequestIdThresholdToDB() {
        STWorkflowInterceptor.updateRequestIdThreshold();
        Integer requestIdThreshold = STWorkflowInterceptor.getRequestIdThreshold();
        String timeThreshold = STWorkflowInterceptor.getTimeThreshold();
        WeaverConnection connection = null;
        WeaverStatement statement = null;
        try {
            connection = ConnectionPool.getInstance().getConnection();
            statement = (WeaverStatement) connection.createStatement();
            Date nowDate = new Date();
            DateFormat dFormat = new SimpleDateFormat("yyyy-MM-dd");
            String nowDate_string = dFormat.format(nowDate);
            String insertSql = "update " + STDBConstants.DATA_THRESHOLD_TABLE_NAME + " set requestid_threshold="+requestIdThreshold+",createdate_threshold='"+timeThreshold+"',updatetime='"+nowDate_string+"' WHERE module ='workflow' ";
            statement.originExecuteUpdate(insertSql);
        } catch (Exception e) {
            LOGGER.error("STWorkflowDataCleaner updateRequestIdThresholdToDB error:", e);
        } finally {
            if (statement != null) {
                statement.closeNoException();
            }
            if (connection != null) {
                ConnectionPool.getInstance().returnConnection(connection);
            }
        }
    }

    private Set<String> saveClearData(WeaverConnection connection, String clearSql, String tableName, String dataIdColumn) throws Exception {
        Set<String> deleteDataIds = new HashSet<>();
        Long startTime = System.currentTimeMillis();
        String selectClearSQL = "";
        clearSql=clearSql.toLowerCase();
        if (STDBSupport.SQLSERVER.equals(STConfig.DB_TYPE)) {
            selectClearSQL=clearSql.replace("delete top(1000)","select "+dataIdColumn+" ");
        } else {
            selectClearSQL = clearSql.replace("delete ", "select " + dataIdColumn + " ");
        }

        String insertClearSQL = "insert into " + STDBConstants.DATA_CLEAR_TABLE_NAME + " (table_name, data_id_column, data_id, clear_time, synced) values (?, ?, ?, ?, ?)";
        WeaverStatement statement = null;
        WeaverPreparedStatement preparedStatement = null;
        try {
            statement = (WeaverStatement) connection.createStatement();
            preparedStatement = (WeaverPreparedStatement) connection.originPreparedStatement(insertClearSQL);
            ResultSet rs = statement.originExecuteQuery(selectClearSQL);
            Long clearTime = System.currentTimeMillis();
            while (rs.next()) {
                String dataId = rs.getString(dataIdColumn);
                deleteDataIds.add(dataId);

                preparedStatement.setString(1, tableName);
                preparedStatement.setString(2, dataIdColumn);
                preparedStatement.setString(3, dataId);
                preparedStatement.setString(4, clearTime.toString());
                preparedStatement.setString(5, "0");
                preparedStatement.addBatch();
            }
            preparedStatement.executeBatch();
        } catch (Exception e) {
            throw e;
        } finally {
            if (statement != null) {
                statement.closeNoException();
            }
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
        Long endTime = System.currentTimeMillis();
        LOGGER.info("STWorkflowDataCleaner saveClearData:" + (endTime - startTime) + "ms");
        return deleteDataIds;
    }

    // 获取删除SQL
    @Override
    public String getClearSql(String tableName) {
        String timeThreshold = STWorkflowInterceptor.getTimeThreshold();
        Integer requestIdThreshold = STWorkflowInterceptor.getRequestIdThreshold();
        if ("workflow_requestbase".equals(tableName.toLowerCase())) {
            if (STDBSupport.ORACLE.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_REQUESTBASE_DEL_ORACLE, tableName, timeThreshold);
            } else if (STDBSupport.SQLSERVER.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_REQUESTBASE_DEL_SQLSERVER, tableName, tableName, timeThreshold);
            } else if (STDBSupport.MYSQL.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_REQUESTBASE_DEL_MYSQL, tableName, timeThreshold);
            }
        } else if ("workflow_currentoperator".equals(tableName.toLowerCase())) {
            if (STDBSupport.ORACLE.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_CURRENTOPERATOR_DEL_ORACLE, tableName, timeThreshold);
            } else if (STDBSupport.SQLSERVER.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_CURRENTOPERATOR_DEL_SQLSERVER, tableName, timeThreshold);
            } else if (STDBSupport.MYSQL.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_CURRENTOPERATOR_DEL_MYSQL, tableName, timeThreshold);
            }
        } else if ("workflow_requestlog".equals(tableName.toLowerCase())) {
            if (STDBSupport.ORACLE.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_REQUESTLOG_DEL_ORACLE, tableName, timeThreshold);
            } else if (STDBSupport.SQLSERVER.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_REQUESTLOG_DEL_SQLSERVER, tableName, timeThreshold);
            } else if (STDBSupport.MYSQL.equals(STConfig.DB_TYPE)) {
                return String.format(WORKFLOW_REQUESTLOG_DEL_MYSQL, tableName, timeThreshold);
            }
        }

        return null;
    }


}
